package org.gl.rtsp;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean;

import org.gl.rtsp.cache.Cache;
import org.gl.rtsp.cache.MemeryCache;
import org.gl.util.ByteUtil;
/**
 * RTSP客户端实现
 * @author Administrator
 *
 */
public class RTSPClient extends Thread implements IEvent {

	private static final String VERSION = " RTSP/1.0\r\n";
	private static final String RTSP_OK = "RTSP/1.0 200 OK";

	/** */
	/** Զ�̵�ַ */
	private final InetSocketAddress remoteAddress;

	/** */
	/** * ���ص�ַ */
	private final InetSocketAddress localAddress;

	/** */
	/** * ����ͨ�� */
	private SocketChannel socketChannel;

	/** */
	/** ���ͻ����� */
	private final ByteBuffer sendBuf;

	/** */
	/** ���ջ����� */
	private final ByteBuffer receiveBuf;

	private static final int BUFFER_SIZE = 2*1024*1024;

	/** */
	/** �˿�ѡ���� */
	private Selector selector;

	private String address;

	private Status sysStatus;

	private String sessionid;

	/** */
	/** �߳��Ƿ����ı�־ */
	private AtomicBoolean shutdown;

	private int seq = 1;

	private boolean isSended;

	private String trackInfo;

	private enum Status {
		init, options, describe, setup, play, pause, teardown
	}
	private Cache cache;
	public RTSPClient(InetSocketAddress remoteAddress,
			InetSocketAddress localAddress, String address) {
		this.remoteAddress = remoteAddress;
		this.localAddress = localAddress;
		this.address = address;

		// ��ʼ��������
		sendBuf = ByteBuffer.allocateDirect(BUFFER_SIZE);
		receiveBuf = ByteBuffer.allocateDirect(BUFFER_SIZE);
		if (selector == null) {
			// �����µ�Selector
			try {
				selector = Selector.open();
			} catch (final IOException e) {
				e.printStackTrace();
			}
		}
		cache  = new MemeryCache();
		startup();
		sysStatus = Status.init;
		shutdown = new AtomicBoolean(false);
		isSended = false;
	}

	public void startup() {
		try {
			// ��ͨ��
			socketChannel = SocketChannel.open();
			// �󶨵����ض˿�
			socketChannel.socket().setSoTimeout(5000);
			socketChannel.configureBlocking(false);
			socketChannel.socket().bind(localAddress);
			if (socketChannel.connect(remoteAddress)) {
				System.out.println("remoteAddress:" + remoteAddress);
			}
			socketChannel.register(selector, SelectionKey.OP_CONNECT
					| SelectionKey.OP_READ | SelectionKey.OP_WRITE, this);

		} catch (final IOException e1) {
			e1.printStackTrace();
		}
	}

	public void send(byte[] out) {
		if (out == null || out.length < 1) {
			return;
		}
		synchronized (sendBuf) {
			sendBuf.clear();
			sendBuf.put(out);
			sendBuf.flip();
		}

		// ���ͳ�ȥ
		try {
			write();
			isSended = true;
		} catch (final IOException e) {
			e.printStackTrace();
		}
	}

	public void write() throws IOException {
		if (isConnected()) {
			try {
				socketChannel.write(sendBuf);
			} catch (final IOException e) {
			}
		} else {
			System.out.println("ͨ��Ϊ�ջ���û��������");
		}
	}

	public byte[] recieve() {
		if (isConnected()) {
			try {
				int len = 0;
				int readBytes = 0;

				synchronized (receiveBuf) {
					receiveBuf.clear();
					try {
						System.out
								.println("----------socketChannelRead-------");
						while ((len = socketChannel.read(receiveBuf)) > 0) {
							readBytes += len;
						}
					} finally {
						receiveBuf.flip();
					}
					if (readBytes > 0) {
						final byte[] tmp = new byte[readBytes];
						receiveBuf.get(tmp);
						// System.out.println("response:" +new String(tmp));
						return tmp;
					} else {
						System.out.println("rceive return <0");
						return null;
					}
				}
			} catch (final IOException e) {
				e.printStackTrace();
				System.out.println("������Ϣ����:");
			}
		} else {
			System.out.println("�˿�û������");
		}
		return null;
	}

	public boolean isConnected() {
		return socketChannel != null && socketChannel.isConnected();
	}

	private void select() {
		int n = 0;
		try {
			if (selector == null) {
				return;
			}
			n = selector.select(1000);// select Ƶ�� ����ָÿ��һ��

		} catch (final Exception e) {
			e.printStackTrace();
		}

		// ���select���ش���0�������¼�
		if (n > 0) {
			for (final Iterator<SelectionKey> i = selector.selectedKeys()
					.iterator(); i.hasNext();) {
				// �õ���һ��Key
				final SelectionKey sk = i.next();
				i.remove();
				// ������Ƿ���Ч
				if (!sk.isValid()) {
					continue;
				}

				// �����¼�
				final IEvent handler = (IEvent) sk.attachment();
				try {
					if (sk.isConnectable()) {
						handler.connect(sk);
					} else if (sk.isReadable()) {
						handler.read(sk);
					} else {
						// System.err.println("Ooops");
					}
				} catch (final Exception e) {
					handler.error(e);
					sk.cancel();
				}
			}
		}
	}

	public void shutdown() {
		if (isConnected()) {
			try {
				socketChannel.close();
				System.out.println("�˿ڹرճɹ�");
			} catch (final IOException e) {
				System.out.println("�˿ڹرմ���:");
			} finally {
				socketChannel = null;
			}
		} else {
			System.out.println("ͨ��Ϊ�ջ���û������");
		}
	}

	@Override
	public void run() {
		// ������ѭ������
		while (!shutdown.get()) {
			try {
				if (isConnected() && (!isSended)) {
					switch (sysStatus) {
					case init:
						doOption();
						break;
					case options:
						doDescribe();
						break;
					case describe:
						doSetup();
						break;
					case setup:
						if (sessionid == null && sessionid.length() > 0) {
							System.out.println("setup��û�����");
						} else {
							doPlay();
						}
						break;
					case play:
						// doPause(); FIXME
						break;

					case pause:
						doTeardown();
						break;
					default:
						break;
					}
				}
				// do select
				select();
				try {
					Thread.sleep(1000);
				} catch (final Exception e) {
				}
			} catch (final Exception e) {
				e.printStackTrace();
			}
		}

		shutdown();
	}

	public void connect(SelectionKey key) throws IOException {
		if (isConnected()) {
			return;
		}
		// ���SocketChannel������
		socketChannel.finishConnect();
		while (!socketChannel.isConnected()) {
			try {
				Thread.sleep(300);
			} catch (final InterruptedException e) {
				e.printStackTrace();
			}
			socketChannel.finishConnect();
		}

	}

	public void error(Exception e) {
		e.printStackTrace();
	}

	public void read(SelectionKey key) throws IOException {
		// ������Ϣ
		final byte[] msg = recieve();
		if (msg != null) {
			handle(msg);
		} else {
			key.cancel();
		}
	}

	private void handle(byte[] msg) {
		String tmp = new String(msg);
		System.out.println("response:sysStatus = " + sysStatus);
		if (tmp.startsWith(RTSP_OK)) {
			switch (sysStatus) {
			case init:
				sysStatus = Status.options;
				break;
			case options:
				sysStatus = Status.describe;
				trackInfo = tmp.substring(tmp.indexOf("trackID"));
				break;
			case describe:
				sessionid = getSession(tmp);
				// sessionid = tmp.substring(tmp.indexOf("Session: ") + 9, tmp
				// .indexOf("Date:"));
				if (sessionid != null && sessionid.length() > 0) {
					sysStatus = Status.setup;
				}
				break;
			case setup:
				sysStatus = Status.play;
				System.out.println("======================");
				System.out.println("=========playing======");
				System.out.println("======================");
				// TODO rtp包解析
				int RTPPacketIndex = 0;
				for (byte b : msg) {
					RTPPacketIndex++;
					if ((char) b == '\r') {
					} else if (msg[RTPPacketIndex + 1] == '\n'
							&& msg[RTPPacketIndex + 2] == '\r'
							&& msg[RTPPacketIndex + 3] == '\n') {
						System.out
								.println("========rtsp packet parse complete===========");
						RTPPacketIndex += 4;
						System.err
								.println("RTPPacketIndex : " + RTPPacketIndex);
						parseRTP(msg, RTPPacketIndex);
						break;
					}
				}
				break;
			case play:
				sysStatus = Status.pause;
				break;
			case pause:
				sysStatus = Status.teardown;
				shutdown.set(true);
				break;
			case teardown:
				sysStatus = Status.init;
				break;
			default:
				break;
			}
			isSended = false;

			System.out.println("response=" + tmp);
		} else {
			if (sysStatus == Status.setup) {
				sysStatus = Status.play;
			}
			System.out.println("steaming");
		}

	}

	private void parseRTP(byte[] msg, int offset) {
		int rtpIndexOffset = offset;
		int rtpPacketLength=0;
		if (isInterleavedTransmission) {
			rtpPacketLength = parseInterleavedTransmission(msg,offset);
			rtpIndexOffset +=4;
		}
		int pt = msg[rtpIndexOffset+2]&0x7f; 
		System.out.println("m : "+(msg[rtpIndexOffset+2]&0x80));
		System.out.println("pt before: "+msg[rtpIndexOffset+2]);
		System.out.println("pt: "+pt);
		cache.cache(msg, rtpIndexOffset, rtpPacketLength);
	}
	/**
	 * //RTSP/RTP 交错传输方式.  
	 */
	final boolean isInterleavedTransmission = true;
	private int parseInterleavedTransmission(byte[] msg, int offset) {
			System.out.println("0x24: " + Integer.toHexString(msg[offset + 0])
					+ " ");
			System.out.println("channel: "
					+ Integer.toHexString(msg[offset + 1]) + " ");
			System.out.println("content length : "
					+ ByteUtil.byteArrayToShort(msg, offset + 2) + " ");
			return  ByteUtil.byteArrayToShort(msg, offset + 2);
	}

	private void doTeardown() {
		StringBuilder sb = new StringBuilder();
		sb.append("TEARDOWN ");
		sb.append(this.address);
		sb.append("/");
		sb.append(VERSION);
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("User-Agent: RealMedia Player HelixDNAClient/10.0.0.11279 (win32)\r\n");
		sb.append("Session: ");
		sb.append(sessionid);
		sb.append("\r\n");
		send(sb.toString().getBytes());
		System.out.println(sb.toString());
	}

	private void doPlay() {
		StringBuilder sb = new StringBuilder();
		sb.append("PLAY ");
		sb.append(this.address);
		sb.append(VERSION);
		sb.append("Session: ");
		sb.append(sessionid);
		sb.append("\r\n");
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("\r\n");
		System.out.println(sb.toString());
		send(sb.toString().getBytes());

	}

	private void doSetup() {
		StringBuilder sb = new StringBuilder();
		sb.append("SETUP ");
		sb.append(this.address + "/trackID=2");// FIXME
												// ��Щ������������IP����ͷʵʱ������Ҫ�ֱ����
		sb.append(VERSION);
		// sb.append("/");
		// sb.append(trackInfo);
		// sb.append("\r\n");
		// sb.append("\r\n");
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("Transport: RTP/AVP/TCP;mode=play\r\n");
		sb.append("\r\n");
		System.out.println(sb.toString());
		send(sb.toString().getBytes());
	}

	private void doOption() {
		System.out.println(this.address);
		StringBuilder sb = new StringBuilder();
		sb.append("OPTIONS ");
		sb.append(address);
		sb.append(VERSION);
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("\r\n");
		System.out.println(sb.toString());
		send(sb.toString().getBytes());
	}

	private void doDescribe() {
		StringBuilder sb = new StringBuilder();
		sb.append("DESCRIBE ");
		sb.append(this.address);
		sb.append(VERSION);
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("\r\n");
		System.out.println(sb.toString());
		send(sb.toString().getBytes());
	}

	private void doPause() {
		StringBuilder sb = new StringBuilder();
		sb.append("PAUSE ");
		sb.append(this.address);
		sb.append("/");
		sb.append(VERSION);
		sb.append("\r\n");
		sb.append("Cseq: ");
		sb.append(seq++);
		sb.append("\r\n");
		sb.append("Session: ");
		sb.append(sessionid);
		sb.append("\r\n");
		send(sb.toString().getBytes());
		System.out.println(sb.toString());
	}

	private static String getSession(String sting) {
		String sessionID = "";
		String key = "Session:";
		String[] lines = sting.split(System.getProperty("line.separator"));
		for (String string : lines) {
			int index = string.indexOf(key);
			if (index >= 0) {
				int index_yinhao = string.indexOf(";");
				if (index_yinhao >= 0) {
					sessionID = string.substring(index + key.length(),
							index_yinhao);
				} else {
					sessionID = string.substring(index + key.length());
				}
				return sessionID.trim();
			}
		}
		return sessionID;
	}

	public static void main(String[] args) {
		try {
			// //目标服务器是
			String url = "rtsp://admin:12345@%s:%d/";
			String ip = "192.168.30.235";
			int port = 554;
			// RTSPClient client = new RTSPClient(new InetSocketAddress(
			// "192.168.2.100", 1935), new InetSocketAddress(
			// "192.168.2.100", 0),
			// "rtsp://192.168.2.100:1935/vod/sample.mp4");

			// RTSPClient client = new RTSPClient(new InetSocketAddress(
			// "127.0.0.1", 5540), new InetSocketAddress(
			// "127.0.0.1", 0),
			// "127.0.0.1:5540/");
			RTSPClient client = new RTSPClient(new InetSocketAddress(ip, port),
					new InetSocketAddress("0.0.0.0", 0), String.format(url, ip,
							port));
			// RTSPClient client = new RTSPClient(new InetSocketAddress(
			// "192.168.2.64", 5540), new InetSocketAddress(
			// "0.0.0.0", 0),
			// "rtsp://192.168.2.64:5540/");

			// RTSPClient client = new RTSPClient(new InetSocketAddress(
			// "192.168.2.100", 8888), new InetSocketAddress(
			// "192.168.2.100", 0),
			// "rtsp://192.168.2.100:8888/");
			client.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}